package cloud.xlh.digital_power_order.service.order.service;

import cloud.xlh.digital_power_order.config.DigitalPowerPayRemoteCall;
import cloud.xlh.digital_power_order.config.rabbitmq.RabbitMQConstant;
import cloud.xlh.digital_power_order.config.rabbitmq.RabbitMQOperateService;
import cloud.xlh.digital_power_order.disruptor.DisruptorMqService;
import cloud.xlh.digital_power_order.disruptor.OrderFactory;
import cloud.xlh.digital_power_order.disruptor.OrderHandler;
import cloud.xlh.digital_power_order.domian.order.OrderSaveDomain;
import cloud.xlh.digital_power_order.domian.order.OrderUpdateDomain;
import cloud.xlh.digital_power_order.enums.OrderStatusEnum;
import cloud.xlh.digital_power_order.exceptionhandle.DPOBusinessExceptionEnum;
import cloud.xlh.digital_power_order.infrastructure.order.entity.OrderEntity;
import cloud.xlh.digital_power_order.infrastructure.order.entity.table.OrderEntityTableDef;
import cloud.xlh.digital_power_order.infrastructure.order.mapper.OrderEntityMapper;
import cloud.xlh.digital_power_order.service.RemoteCallManageService;
import cloud.xlh.digital_power_order.service.order.adapter.OrderServiceAdapter;
import cn.hutool.extra.spring.SpringUtil;
import com.alibaba.fastjson.JSONObject;
import com.lmax.disruptor.EventFactory;
import com.lmax.disruptor.YieldingWaitStrategy;
import com.lmax.disruptor.dsl.Disruptor;
import com.lmax.disruptor.dsl.ProducerType;
import com.mybatisflex.core.query.QueryCondition;
import com.mybatisflex.core.update.UpdateChain;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Service;

import javax.annotation.Resource;
import java.net.SocketTimeoutException;
import java.util.HashMap;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Executors;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.function.BiConsumer;
import java.util.function.Function;
import java.util.function.Supplier;

/**
 * Created with IntelliJ IDEA.
 *
 * @ project name : xlh_osms
 * @ Author: XuLeHuang
 * @ Date: 2023/8/12 18:53
 * @ Description:
 */
@Slf4j
@Service
public class OrderService {

    @Resource
    private OrderServiceAdapter orderServiceAdapter;
    @Resource
    private OrderEntityMapper orderEntityMapper;
    @Resource
    private RemoteCallManageService remoteCallManageService;
    EventFactory<OrderEntity> factory = new OrderFactory();
    private int ringBufferSize = 1024 * 1024;
    private Disruptor<OrderEntity> disruptor =
            new Disruptor<OrderEntity>(factory, ringBufferSize, Executors.newSingleThreadExecutor(), ProducerType.SINGLE, new YieldingWaitStrategy());
    /*
     * 该方法传入的消费者需要实现WorkHandler接口，方法的内部实现是：先创建WorkPool，然后封装WorkPool为EventHandlerPool返回。
     * 消费者1、2对于消息的消费有时有竞争，保证同一消息只能有一个消费者消费
     */
    {
        disruptor.handleEventsWithWorkerPool(new OrderHandler("1"), new OrderHandler("2"));
    }

    @Resource
    private DisruptorMqService disruptorMqService;

    @Resource
    private RabbitMQOperateService rabbitMQOperateService;

    public String genOrder(OrderSaveDomain orderSaveDomain) {

        /*disruptor.start();
        RingBuffer<OrderEntity> ringBuffer = disruptor.getRingBuffer();
        Producer producer = new Producer(ringBuffer);
        //单生产者，生产3条数据
        for (int l = 0; l < 3; l++) {
            producer.onData(orderSaveDomain);
        }
        //为了保证消费者线程已经启动，留足足够的时间。具体原因详见另一篇博客：disruptor的shutdown失效问题
        try {
            Thread.sleep(1000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        //会阻塞，知道所有事件被处理
        disruptor.shutdown();*/
        try {
            disruptorMqService.genOrderMessage(orderSaveDomain);
        } catch (Exception e) {
            throw DPOBusinessExceptionEnum.DISRUPTOR_EXCEPTION.newException(e.getCause());
        }

        return "000000";
    }

    public String selectOrder(String orderNo) {
        try {

            DigitalPowerPayRemoteCall digitalPowerPayRemoteCall = (DigitalPowerPayRemoteCall) remoteCallManageService.getRemoteCallService("digitalPowerPayRemoteCall");
            String result = digitalPowerPayRemoteCall.selectOrder(orderNo);
        } catch (Exception e) {
            Class<? extends Throwable> aClass = e.getCause().getClass();
            if (SocketTimeoutException.class.getName().equals(aClass.getName())) {
                log.error("超时了");
            }
            e.printStackTrace();
        }
        return "000000";
    }

    public String payOrder(String orderNo) {
        ThreadPoolExecutor iOThreadPoolExecutor = SpringUtil.getBean("IOThreadPoolExecutor");
        //异步支付，等待支付结果
        //超时
        CompletableFuture<JSONObject> payOrderCompletableFuture = CompletableFuture.supplyAsync(new Supplier<JSONObject>() {
            @Override
            public JSONObject get() {
                DigitalPowerPayRemoteCall remoteCallService = (DigitalPowerPayRemoteCall)remoteCallManageService.getRemoteCallService(DigitalPowerPayRemoteCall.class.getSimpleName());
                JSONObject req = new JSONObject();
                remoteCallService.payOrder(req);
                return req;
            }
        }, iOThreadPoolExecutor).whenComplete(new BiConsumer<JSONObject, Throwable>() {
            @Override
            public void accept(JSONObject jsonObject, Throwable throwable) {
                UpdateChain.of(OrderEntity.class)
                        .set(OrderEntity::getStatus, OrderStatusEnum.PAY_SUCCESS.getCode())
                        .where(OrderEntity::getOrderNo).eq(orderNo)
                        .update();
                jsonObject.put("orderStatus" , OrderStatusEnum.PAY_SUCCESS.getCode());
                jsonObject.put("orderPayResult" , OrderStatusEnum.PAY_SUCCESS.getDesc());
            }
        }).exceptionally(new Function<Throwable, JSONObject>() {
            @Override
            public JSONObject apply(Throwable throwable) {
                if (SocketTimeoutException.class.getName().equals(throwable.getClass().getName())) {
                    log.error("超时了");
                    //查询一次
                    HashMap<String, String> parmaMap = new HashMap<>();
                    parmaMap.put("exchange" , RabbitMQConstant.COMMON_ORDER_EXCHANGE_NAME);
                    parmaMap.put("routingKey" , RabbitMQConstant.ORDER_BACK_QUERY_ROUTING_KEY);
                    parmaMap.put("message" , orderNo);
                    //10秒过期
                    parmaMap.put("ttl" , "10000");
                    rabbitMQOperateService.process(parmaMap);
                }
                JSONObject result = new JSONObject();
                result.put("orderStatus" , "");
                return result;
            }
        }).whenCompleteAsync(new BiConsumer<JSONObject, Throwable>() {
            @Override
            public void accept(JSONObject jsonObject, Throwable throwable) {
                //更新数据库
                String orderStatus = (String) jsonObject.get("orderStatus");
                UpdateChain.of(OrderEntity.class)
                        .set(OrderEntity::getStatus, OrderStatusEnum.PAY_SUCCESS.getCode())
                        .where(OrderEntity::getOrderNo).eq(orderNo)
                        .update();
            }
        });

        CompletableFuture<JSONObject> payResultNotifyCompletableFuture = CompletableFuture.supplyAsync(new Supplier<JSONObject>() {
            @Override
            public JSONObject get() {
                OrderEntity orderEntity = null;
                //查本地数据库
                int limit = 3;
                long unit = 1000l;
                for (int i = 0; i < limit; i++) {
                    try {
                        Thread.sleep((i + 1) * unit);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                    //调用数据库查询
                    orderEntity = orderEntityMapper.selectOneById(orderNo);
                    String status = orderEntity.getStatus();
                    if (OrderStatusEnum.PAY_EXCEPTION.getCode().equals(status) ||
                        OrderStatusEnum.PAY_FAIL.getCode().equals(status) ||
                        OrderStatusEnum.PAY_SUCCESS.getCode().equals(status)
                    ) {
                        UpdateChain.of(OrderEntity.class)
                                .set(OrderEntity::getStatus, status)
                                .where(OrderEntity::getOrderNo).eq(orderNo)
                                .update();
                        break;
                    }
                }
                JSONObject result = new JSONObject();
                result.put("orderStatus" , orderEntity.getStatus());
                return result;
            }
        }, iOThreadPoolExecutor);
        CompletableFuture<JSONObject> resultNotifyCompletableFuture  = (CompletableFuture<JSONObject>)CompletableFuture.anyOf(payOrderCompletableFuture, payOrderCompletableFuture).join();
        ;
        try {
            JSONObject result = resultNotifyCompletableFuture.get();
        } catch (InterruptedException e) {
            e.printStackTrace();
        } catch (ExecutionException e) {
            e.printStackTrace();
        }
        return "等待支付结果中......";
    }

    public String refundOrder(OrderUpdateDomain orderUpdateDomain) {
        OrderEntity orderEntity = orderServiceAdapter.convertOrderEntity(orderUpdateDomain);
        orderEntity.setStatus(String.valueOf(OrderStatusEnum.REFUND_APPLYING.getCode()));
        orderEntityMapper.updateByCondition(orderEntity
                , QueryCondition.create(OrderEntityTableDef.ORDER_ENTITY.ORDER_NO, orderEntity.getOrderNo()));
        DigitalPowerPayRemoteCall remoteCallService = (DigitalPowerPayRemoteCall)remoteCallManageService.getRemoteCallService(DigitalPowerPayRemoteCall.class.getSimpleName());
        String result = remoteCallService.refundOrder(orderUpdateDomain);
        return "退款申请中......";
    }
}
